 1.Spark原理之数据倾斜下基本概念
   
   这是我们期望的处理模式：
   什么是数据倾斜
   Task之间数据分配的非常不均匀
   key.hashCode % reduce个数 = 分区号
   数据倾斜有哪些现象
       Executor lost、OOM、Shuffle过程出错、程序执行慢
       单个Executor执行时间特别久，整体任务卡在某个阶段不能结束
       正常运行的任务突然失败大多数 Task 运行正常，个别Task运行缓慢或发生OOM
   
   数据倾斜造成的危害有哪些
       个别任务耗时远高于其它任务，轻则造成系统资源的浪费，使整体应用耗时过大，不能充分发挥分布式系统并
行计算的优势
       个别Task发生OOM，导致整体作业运行失败
   
   为什么会发生数据倾斜
   数据异常
       参与计算的 key 有大量空值(null)，这些空值被分配到同一分区
   
   Map Task数据倾斜，主要是数据源导致的数据倾斜：
       数据文件压缩格式（压缩格式不可切分）
       Kafka数据分区不均匀
   Reduce task数据倾斜（重灾区，最常见）：
       Shuffle （外因）。Shuffle操作涉及到大量的磁盘、网络IO，对作业性能影响极大
       Key分布不均 （内因）
   
   如何定位发生数据倾斜
   凭借经验或Web UI，找到对应的Stage；再找到对应的 Shuffle 算子
 
 2.数据倾斜处理
   
   做好数据预处理：
       过滤key中的空值
       消除数据源带来的数据倾斜(文件采用可切分的压缩方式)
   数据倾斜产生的主要原因：Shuffle + key分布不均
   
   处理数据倾斜的基本思路：
       消除shuffle
       减少shuffle过程中传输的数据
       选择新的可用于聚合或join的Key(结合业务)
       重新定义分区数
       加盐强行打散Key
   key.hashCode % reduce个数 = 分区号
   1).避免shuffle
          Map端的join是典型的解决方案
          可以完全消除Shuffle，进而解决数据倾斜
          有很强的适用场景(大表和小表关联)，典型的大表与小表的join，其他场景不合适
   2).减少 Shuffle 过程中传输的数据
          使用高性能算子，避免使用groupByKey，用reduceByKey或aggregateByKey替代
          没有从根本上解决数据分配不均的问题，收效有限，使用场景有限
   3).选择新的可用于聚合或join的Key
          从业务出发，使用新的key去做聚合或join。如当前key是【省 城市 日期】,在业务允许的情况下选择新的
key【省 城市 区 日期】,有可能 解决或改善 数据倾斜问题
          存在的问题：这样的key不好找；或者找到了新的key也不能解决问题
   4).改变Reduce的并行度
   key.hashCode % reduce个数 = 分区号
          变更 reduce 的并行度。理论上分区数从 N 变为 N-1 有可能解决或改善数据倾斜
		  一般情况下这个方法不管用,数据倾斜可能是由很多key造成的,但是建议试试因为这个方法非常简单,成本极低
		  可能只是解决了这一次的数据倾斜问题，非长远之计
          缺点：适用性不广；优点：简单
// 2000W条数据对Key做了处理，使其在后面的shuffle中产生数据倾斜
val rdd = sc.makeRDD(1 to 30000000)
val rdd1 = rdd.map(x => (if (x>3000000) (x%3000000)*6 else x, 1))

// 执行过程中可以感觉到有一个作业执行的慢，从Web界面中可以很清楚的看见数据倾斜
// 备注：shuffle write是均衡的，shuffle read不均衡
rdd1.groupByKey().mapPartitionsWithIndex{(index, iter)=>
val elementCount = iter.toArray.length
Iterator(index + ":" + elementCount)
}.collect

// 调整(缩小)了并行度,增加了shuffle,但执行的时间更快了。数据倾斜猛于虎,在这里数据倾斜对系统的影响超过了
// shuffle
rdd1.repartition(5).groupByKey().mapPartitionsWithIndex{(index, iter)=>
val elementCount = iter.toArray.length
Iterator(index + ":" + elementCount)
}.collect

// repartition在重分区的过程中会产生shuffle,而coalesce不会,这样可以减少一次shuffle,进而提高效率
rdd1.coalesce(5).groupByKey().mapPartitionsWithIndex{(index, iter)=>
val elementCount = iter.toArray.length
Iterator(index + ":" + elementCount)
}.collect

// 定义分区器，减少了一次shuffle，执行速度更快
rdd1.groupByKey(new org.apache.spark.HashPartitioner(5)).mapPartitionsWithIndex{(index,
iter)=>
val elementCount = iter.toArray.length
Iterator(index + ":" + elementCount)
}.collect
   5).加盐强行打散Key
   shuffle + key不能分散
   (1).两阶段聚合
   加盐打散key。给每个key都加一个随机数，如10以内的随机数。此时key就被打散了
   局部聚合。对打上随机数的数据，执行一次聚合操作，得到结果
   全局聚合。将各个key的前缀去掉，再进行一次聚合操作，得到最终结果
package cn.lagou.advanced

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

// 两阶段聚合解决数据倾斜
object DataSkew2 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("").setMaster("local[2]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    // 本地文件是32M一个块；文件大小为88M，如果不做重分区那么将有3个数据块
    val rdd1: RDD[String] = 
      sc.textFile("data/log1.txt").coalesce(2)
    println(s"rdd1.Partitions = ${rdd1.getNumPartitions}")

    val rdd2 = rdd1.flatMap(_.split("\\s+"))
      .map((_, 1))
      .groupByKey()
      .map{case (k, v) => (k, v.sum)}
    println(s"rdd2.Partitions = ${rdd2.getNumPartitions}")
    rdd2.collect.foreach(println)

    // 可以发现有数据倾斜
    val info = getRDDPartitionCount(rdd2)
    println(info)

    // 使用两阶段聚合，解决数据倾斜问题
    // 加前缀，执行局部聚合；去前缀，执行全局聚合
    val random = scala.util.Random
    val rdd3: RDD[(String, Int)] = rdd1.filter(_.trim.size > 0)
      .flatMap(_.split("\\s+"))
      .map { word =>
        val prefix = random.nextInt(5)
        (s"${prefix}_$word", 1)
      }.groupByKey()
      .map { case (k, v) =>
        val word: String = k.split("_")(1)
        (word, v.sum)
      }.groupByKey()
      .mapValues(v => v.sum)
    rdd3.collect().foreach(println)

    println("end!")
    Thread.sleep(100000000)
    sc.stop()
  }

  def getRDDPartitionCount(rdd: RDD[_]): String = {
    rdd.mapPartitionsWithIndex{case (idx, iter) =>
      Iterator(s"$idx : ${iter.size}")
    }.collect().mkString("\n")
  }
}   
   两阶段聚合的优缺点：
       对于聚合类的shuffle操作导致的数据倾斜，效果不错。通常都可以解决掉数据倾斜,至少是大幅度缓解数据倾
斜，将Spark作业的性能提升数倍以上
       仅适用于聚合类的shuffle操作，适用范围相对较窄。如果是join类的shuffle操作,还得用其他的解决方案
   (2).采样倾斜key并拆分join操作
   业务场景：两个RDD/两张表进行 join 的时候，数据量都比较大。
   使用场景：计算两个RDD/两张表中的key分布情况。如果出现数据倾斜，是其中一个RDD/Hive表中的少数几个key的
数据量过大，而另一个RDD/Hive表中的所有key都分布比较均匀，那么采用这个解决方案比较合适。
   处理步骤：
   1)、对包含少数几个数据量过大的key的那个RDD，通过sample算子采样出一份样本来，然后统计一下每个key的数
量，计算出数据量最大的是哪几个key；
   2)、将这几个key对应的数据从原来的RDD中拆分出来，形成一个单独的RDD，并给每个key都打上n以内的随机数作
为前缀，而不会导致倾斜的大部分key形成另外一个RDD；
   3)、将需要join的另一个RDD，也过滤出来那几个倾斜key对应的数据并形成一个单独的RDD，将每条数据膨胀成n条
数据，这n条数据都按顺序附加一个0~n的前缀，不会导致倾斜的大部分key也形成另外一个RDD；
   4)、再将附加了随机前缀的独立RDD与另一个膨胀n倍的独立RDD进行join，此时就可以将原先相同的key打散成n
份，分散到多个task中去进行join了；
   5)、另外两个普通的RDD就照常join即可；
   6)、最后将两次join的结果使用union算子合并起来即可，就是最终的join结果。
   (3).使用随机前缀和扩容再进行join
   业务场景：如果在进行join操作时，RDD中有大量的key导致数据倾斜，进行分拆key没什么意义，此时就只能使用最
后一种方案来解决问题了。
   处理步骤：
   1)、选一个RDD，将每条数据都打上一个n以内的随机前缀(打散)
   2)、对另外一个RDD进行扩容,将每条数据都扩容成n条数据,扩容出来的每条数据都依次打上一个0~n的前缀
   3)、将两个处理后的RDD进行join即可
   优缺点：
       如果两个RDD都很大，那么将RDD进行N倍的扩容显然行不通
       使用扩容的方式通常能缓解数据倾斜，不能彻底解决数据倾斜问题
   
   小结：
   数据倾斜问题的解决没有银弹，通常是找到数据倾斜发生的原因，然后见招拆招；
   在实践中很多情况下，如果只是处理较为简单的数据倾斜场景，使用上述方案中的某一种基本就可以解决。
   但是如果要处理一个较为复杂的数据倾斜场景,可能需要将多种方案组合起来使用。
   需要对这些方案的思路和原理都透彻理解之后,在实践中根据各种不同的情况,灵活运用多种方案,来解决自己遇到
的数据倾斜问题。